package hadoop.mapreduce.reducerjoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * 相当于一个MapTask
 */
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {

    private String fileName;

    /**
     * 默认使用FileInputFormat进行切片,按文件切片,切片大小默认等于切块Block大小(128M),所以会有两个切片
     * 一个切片执行一个MapTask
     * TextInputFormat是FileInputFormat的默认实现,该实现按行读取文件,每读取一行数据执行一次map方法
     * @param key 偏移量
     * @param value 一行数据
     * @param context
     * @throws IOException
     * @throws InterruptedException
     *输入数据
     * id    pid  amount
     * 1001	 01	  1
     *
     * pid  pName
     * 01  小米
     *
     *为啥输出的key为pid,因为两个文件都有
     *
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        TableBean tableBean = new TableBean();
        String[] split = value.toString().split("\t");
        if (fileName.contains("order")){
            tableBean.setId(split[0]);
            tableBean.setPid(split[1]);
            tableBean.setAmount(Long.parseLong(split[2]));
            tableBean.setpName("");
            tableBean.setFlag("order");
            context.write(new Text(split[1]),tableBean);
        }else {
            tableBean.setId("");
            tableBean.setPid(split[0]);
            tableBean.setAmount(0);
            tableBean.setpName(split[1]);
            tableBean.setFlag("pd");
            context.write(new Text(split[0]),tableBean);
        }
    }

    /**
     * 此方法在MapTask任务开始前执行一次,此处用来获取文件名
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit fileSplit = (FileSplit)context.getInputSplit();
        fileName = fileSplit.getPath().getName();
    }

}
